-- Copyright (c) 2024 Huawei Technologies Co., Ltd.
-- openUBMC is licensed under Mulan PSL v2.
-- You can use this software according to the terms and conditions of the Mulan PSL v2.
-- You may obtain a copy of Mulan PSL v2 at:
--         http://license.coscl.org.cn/MulanPSL2
-- THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
-- EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
-- MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
-- See the Mulan PSL v2 for more details.

local lu = require 'luaunit'
local sml = require 'sml'
local drives_object = require 'drives.drives_object'
local c_pd_info = require 'sml.pd_info'
local c_pd_sas_smart_info = require 'sml.pd_sas_smart_info'
local pd_object = require 'pd.pd_object'
local STOP_LOCATE <const> = 0
local LOCATE <const> = 1
local NUMBER_DRIVE_NAME = 1                -- 非法数字格式的drive_name
local INVALID_LOCATION_INDICATOR_STATE = 2 -- 非法的location_indicator_state
local file_sec = require 'utils.file'
local utils = require 'mc.utils'
local project_dir = os.getenv('PROJECT_DIR')
local add_event = require 'add_event'
local client = require 'storage.client'
local storageApp = require 'storage_app'
local c_nvme = require 'nvme.nvme_object'
local method_misc = require 'method_misc'
local mdb = require 'mc.mdb'
local rpc_service_subhealth = require 'rpc_services.rpc_service_subhealth'
local rpc_service_drive = require 'rpc_services.rpc_service_drive'
local common_def = require 'common_def'

local function make_pd_info(test_data)
    return c_pd_info.new({
        model = test_data.pd_model,
        temperature = test_data.pd_temperature,
        coerced_size = test_data.pd_capacity,
        interface_type = test_data.pd_interface_type,
        device_speed = test_data.pd_device_speed,
        link_speed = test_data.pd_link_speed,
        media_type = test_data.pd_media_type,
        fw_state = test_data.pd_fw_status,
        serial_num = test_data.pd_sn,
        manufacturer = test_data.pd_manufacturer,
        power_on_hours = test_data.pd_ssd_power_on_hours,
        firmware_version = test_data.pd_fw_version,
        rotation_speed = test_data.pd_rotation_speed,
        block_size = test_data.pd_block_size,
        sas_addr1 = test_data.pd_sas_addr_1,
        sas_addr2 = test_data.pd_sas_addr_2,
        remnant_media_wearout = test_data.pd_ssd_lifeleft,
        prefail_count = test_data.pd_predictive_fail_count,
        media_err_count = test_data.pd_media_err_count,
        other_err_count = test_data.pd_other_err_count,
        proginfo = {
            patrol_state = test_data.pd_patrol_status,
            rebuild_state = test_data.pd_rebuild_state,
            rebuild_progress = test_data.pd_rebuild_progress
        },
        power_state = test_data.pd_power_state,
        boot_priority = test_data.pd_boot_priority,
        hot_spare = test_data.pd_hot_spare,
        last_prefail_event_seq_num = test_data.pd_last_prefail_event_seq_num
    })
end

local function construct_ctx()
    local ctx = {}
    ctx.ChanType = 1
    ctx.get_initiator = function()
        return {}
    end

    return ctx
end

local ctx = construct_ctx()

local function make_pd_sas_smart_info(test_data)
    return c_pd_sas_smart_info.new({
        strip_temperature = test_data.strip_temperature,
        glist_len = test_data.glist_len,
        plist_len = test_data.plist_len,
        manufacture_data = test_data.manufacture_data,
        blocks_sent = test_data.blocks_sent,
        blocks_received = test_data.blocks_received,
        minutes_left = test_data.minutes_left,
    })
end

local function update_test_data(test_data)
    test_data.pd_model = 'new_model'
    test_data.pd_temperature = test_data.pd_temperature - 1
    test_data.pd_capacity = test_data.pd_capacity + 1
    test_data.pd_interface_type = 'new_SAS'
    test_data.pd_device_speed = 3
    test_data.pd_link_speed = 3
    test_data.pd_media_type = 2
    test_data.pd_fw_status = 2
    test_data.pd_sn = 'NEWZC1ALLK4'
    test_data.pd_manufacturer = 'NEWATA'
    test_data.pd_ssd_power_on_hours = 255
    test_data.pd_fw_version = 'NEWTN05'
    test_data.pd_rotation_speed = 7000
    test_data.pd_block_size = 612
    test_data.pd_sas_addr_1 = 'NEW5942533A44D5E000'
    test_data.pd_sas_addr_2 = 'NEWN/A'
    test_data.pd_ssd_lifeleft = 15
    test_data.pd_predictive_fail_count = 52
    test_data.pd_media_err_count = 15
    test_data.pd_other_err_count = 66
    test_data.pd_patrol_status = 2
    test_data.pd_power_state = 1
    test_data.pd_rebuild_state = 1
    test_data.pd_rebuild_progress = 30
    test_data.pd_boot_priority = 2
    test_data.pd_hot_spare = 0
    test_data.pd_last_prefail_event_seq_num = 123
end

local function check_drive(drive, test_data)
    lu.assertEquals(drive.Model, test_data.pd_model)
    lu.assertEquals(drive.TemperatureCelsius, test_data.pd_temperature)
    lu.assertEquals(drive.CapacityMiB, test_data.pd_capacity)
    lu.assertEquals(drive.Protocol, test_data.pd_interface_type)
    lu.assertEquals(drive.CapableSpeedGbs, test_data.pd_device_speed)
    lu.assertEquals(drive.NegotiatedSpeedGbs, test_data.pd_link_speed)
    lu.assertEquals(drive.MediaType, test_data.pd_media_type)
    lu.assertEquals(drive.FirmwareStatus, test_data.pd_fw_status)
    lu.assertEquals(drive.SerialNumber, test_data.pd_sn)
    lu.assertEquals(drive.Manufacturer, test_data.pd_manufacturer)
    lu.assertEquals(drive.PowerOnHours, test_data.pd_ssd_power_on_hours)
    lu.assertEquals(drive.Revision, test_data.pd_fw_version)
    lu.assertEquals(drive.RotationSpeedRPM, test_data.pd_rotation_speed)
    lu.assertEquals(drive.BlockSizeBytes, test_data.pd_block_size)
    lu.assertEquals(drive.SASAddress1, test_data.pd_sas_addr_1)
    lu.assertEquals(drive.SASAddress2, test_data.pd_sas_addr_2)
    lu.assertEquals(drive.PredictedMediaLifeLeftPercent, test_data.pd_ssd_lifeleft)
    lu.assertEquals(drive.PredictedFailCount, test_data.pd_predictive_fail_count)
    lu.assertEquals(drive.MediaErrorCount, test_data.pd_media_err_count)
    lu.assertEquals(drive.OtherErrorCount, test_data.pd_other_err_count)
    lu.assertEquals(drive.PatrolState, test_data.pd_patrol_status)
    lu.assertEquals(drive.PowerState, test_data.pd_power_state)
    lu.assertEquals(drive.RebuildState, test_data.pd_rebuild_state)
    lu.assertEquals(drive.RebuildProgress, test_data.pd_rebuild_progress)
    lu.assertEquals(drive.BootPriority, test_data.pd_boot_priority)
    lu.assertEquals(drive.HotspareType, test_data.pd_hot_spare)
    lu.assertEquals(drive.LastPrefailEventSeqNum, test_data.pd_last_prefail_event_seq_num)
end

local function create_drive(self)
    local drive_info = { HddBackplaneStartSlot = 0, Presence = 1, RelativeSlot = 1, ObjectName = 'Drive1' }
    self.object_manager.mc:add_object('Drive', drive_info)
    self.bus_monitor_service.on_os_state_changed:emit(true)
    local drive = self.drive_collection:get_drive(drive_info.Name)
    local pd_info = { controller_id = 1, controller_type = 0 }
    local pd = { slot_num = 2, enclosure_id = 3, device_id = 1 }
    local obj = pd_object.new(pd_info.controller_id, pd, pd_info.controller_type)
    drive:identified(obj)

    return drive_info, drive
end

function TestStorage:test_add_drive_object()
    local _, drive = create_drive(self)
    lu.assertNotIsNil(drive)

    -- 模拟定位 pd
    local pd = { controller_id = 1, slot_num = 2, enclosure_id = 3, device_id = 1 }
    lu.assertIsTrue(drive:is_identified())
    lu.assertEquals(drive.RefControllerId, pd.controller_id)
    lu.assertEquals(drive.SlotNumber, pd.slot_num)
    lu.assertEquals(drive.EnclosureId, pd.enclosure_id)
    lu.assertEquals(drive.device_id, pd.device_id)

    -- hook drive 用到的 sml 函数
    local params = {}

    ---@diagnostic disable-next-line: duplicate-set-field
    sml.get_pd_info = function(...)
        params['get_pd_info'] = { ... } -- 启动任务更新 drive 信息
        return make_pd_info(self.test_pd_info)
    end
    ---@diagnostic disable-next-line: duplicate-set-field
    sml.get_pd_sas_smart_info = function(...)
        params['get_pd_sas_smart_info'] = { ... } -- 启动任务更新 drive 信息
        return make_pd_sas_smart_info(self.test_pd_sas_smart_info)
    end

    drives_object.get_instance().assert_max_temp = function ()
        return false, false
    end

    self.tasks:run_all_task()
    lu.assertEquals(drive:get_task_count(), 5)

    -- 校验任务执行结果
    lu.assertEquals(params, { get_pd_info = { {Priority = 'Secondary'}, pd.controller_id, drive.device_id } })
    lu.assertEquals(drive.NodeId, 'HDDPlaneDisk1')
    check_drive(drive, self.test_pd_info)

    -- 模拟解除 pd 定位，所有任务都会退出
    drive:identified(nil)
    lu.assertEquals(drive:get_task_count(), 2)

    -- 重新定位，模拟换一个 controller_id 但是 slot_num 和 enclosure_id 不变，
    -- 定位后会重启所有任务，这里故意更换一下静态信息，确保静态信息也能得到更新
    local new_pd = { controller_id = 2, slot_num = 2, enclosure_id = 3, device_id = 2 }
    update_test_data(self.test_pd_info)

    drive:identified(new_pd)
    self.tasks:run_all_task()

    lu.assertEquals(params, { get_pd_info = { {Priority = 'Secondary'}, new_pd.controller_id, drive.device_id } })
    check_drive(drive, self.test_pd_info)
end

-- pd_operation 打桩
local function sml_pd_operation(self)
    return function(controller_index, device_id, operation, data)
        local sml_drive = self.drive_collection:get_drive_by_id(controller_index, device_id)
        if operation == STOP_LOCATE then
            sml_drive.LocateLed = STOP_LOCATE
        elseif operation == LOCATE then
            sml_drive.LocateLed = LOCATE
        elseif operation == 5 then                            -- 5: 设置固件状态
            if data == string.pack('B', 0) .. '\x01' then     -- 0: 设置 UG 状态
                sml_drive.FirmwareStatus = 0
            elseif data == string.pack('B', 6) .. '\x01' then -- 6: 设置 online 状态
                sml_drive.FirmwareStatus = 6                  -- 6: 设置 online 状态
            end
        elseif operation == 8 then                            -- 8: 设置启动盘
            if data == string.pack('B', 1) then               -- 1: 设置启动盘
                sml_drive.BootPriority = 1
            elseif data == string.pack('B', 0) then           -- 0 取消启动盘
                sml_drive.BootPriority = 0
            end
        elseif operation == 9 then              -- 9: 设置巡检状态
            if data == string.pack('B', 1) then -- 1: 开启巡检状态
                sml_drive.PatrolState = 1
            elseif data == string.pack('B', 0) then
                sml_drive.PatrolState = 0
            end
        elseif operation == 2 then -- 2: 设置全局热备盘
            sml_drive.HotspareType = 1
        elseif operation == 3 then -- 3: 设置局部热备盘
            sml_drive.HotspareType = 2
        elseif operation == 4 then -- 4 取消热备盘
            sml_drive.HotspareType = 0
        end
        return 0
    end
end

function TestStorage:test_set_drive_location_indicator_state()
    local drive_info, drive = create_drive(self)

    sml.pd_operation = sml_pd_operation(self)

    local rpc_service_drive = self.rpc_service_drive
    self.tasks:run_all_task()

    -- 设置定位灯
    drive.LocateLed = STOP_LOCATE
    local ret = pcall(rpc_service_drive.drive_operate, rpc_service_drive, 'SetLocationIndicatorState', drive_info.Name,
        ctx, LOCATE)
    lu.assertEquals(ret, true)
    lu.assertEquals(drive.LocateLed, LOCATE)
    ret = pcall(rpc_service_drive.drive_operate, rpc_service_drive, 'SetLocationIndicatorState', drive_info.Name, ctx,
        STOP_LOCATE)
    lu.assertEquals(ret, true)
    lu.assertEquals(drive.LocateLed, STOP_LOCATE)
    ret = pcall(rpc_service_drive.drive_operate, rpc_service_drive, 'SetLocationIndicatorState', NUMBER_DRIVE_NAME, ctx,
        STOP_LOCATE)
    lu.assertEquals(ret, false)
    ret = pcall(rpc_service_drive.drive_operate, rpc_service_drive, 'SetLocationIndicatorState', NUMBER_DRIVE_NAME, ctx,
        LOCATE)
    lu.assertEquals(ret, false)
    ret = pcall(rpc_service_drive.drive_operate, rpc_service_drive, 'SetLocationIndicatorState', "Invalid_Param", ctx,
        STOP_LOCATE)
    lu.assertEquals(ret, false)
    ret = pcall(rpc_service_drive.drive_operate, rpc_service_drive, 'SetLocationIndicatorState', "Invalid_Param", ctx,
        LOCATE)
    lu.assertEquals(ret, false)
    ret = pcall(rpc_service_drive.drive_operate, rpc_service_drive, 'SetLocationIndicatorState', drive_info.Name, ctx,
        INVALID_LOCATION_INDICATOR_STATE)
    lu.assertEquals(ret, false)
    ret = pcall(rpc_service_drive.drive_operate, rpc_service_drive, 'SetLocationIndicatorState', drive_info.Name, ctx,
        INVALID_LOCATION_INDICATOR_STATE)
    lu.assertEquals(ret, false)
    drive.RefControllerId = 0xff
    ret = pcall(rpc_service_drive.drive_operate, rpc_service_drive, 'SetLocationIndicatorState', drive_info.Name, ctx,
        LOCATE)
    lu.assertEquals(ret, true)
end

function TestStorage:test_set_drive_fault_indicator_state()
    local drive_info = create_drive(self)

    local rpc_service_drive = self.rpc_service_drive
    self.tasks:run_all_task()

    -- 设置故障灯
    local ret = pcall(rpc_service_drive.drive_operate, rpc_service_drive, 'SetFaultIndicatorState', drive_info.Name,
        ctx, LOCATE)
    lu.assertEquals(ret, true)
    ret = pcall(rpc_service_drive.drive_operate, rpc_service_drive, 'SetFaultIndicatorState', drive_info.Name, ctx,
        STOP_LOCATE)
    lu.assertEquals(ret, true)
    ret = pcall(rpc_service_drive.drive_operate, rpc_service_drive, 'SetFaultIndicatorState', NUMBER_DRIVE_NAME, ctx,
        STOP_LOCATE)
    lu.assertEquals(ret, false)
    ret = pcall(rpc_service_drive.drive_operate, rpc_service_drive, 'SetFaultIndicatorState', NUMBER_DRIVE_NAME, ctx,
        LOCATE)
    lu.assertEquals(ret, false)
    ret = pcall(rpc_service_drive.drive_operate, rpc_service_drive, 'SetFaultIndicatorState', "Invalid_Param", ctx,
        STOP_LOCATE)
    lu.assertEquals(ret, false)
    ret = pcall(rpc_service_drive.drive_operate, rpc_service_drive, 'SetFaultIndicatorState', "Invalid_Param", ctx,
        LOCATE)
    lu.assertEquals(ret, false)
    ret = pcall(rpc_service_drive.drive_operate, rpc_service_drive, 'SetFaultIndicatorState', drive_info.Name, ctx,
        INVALID_LOCATION_INDICATOR_STATE)
    lu.assertEquals(ret, false)
    ret = pcall(rpc_service_drive.drive_operate, rpc_service_drive, 'SetFaultIndicatorState', drive_info.Name, ctx,
        INVALID_LOCATION_INDICATOR_STATE)
    lu.assertEquals(ret, false)
end

function TestStorage:test_set_drive_hotspare_type()
    local drive_info, drive = create_drive(self)

    sml.pd_operation = sml_pd_operation(self)

    local rpc_service_drive = self.rpc_service_drive
    self.tasks:run_all_task()

    drive.HotspareType = 0
    lu.assertEquals(drive.HotspareType, 0)
    sml.get_pd_info = function()
        return { hot_spare = 1 }
    end
    local ret = pcall(rpc_service_drive.drive_operate, rpc_service_drive, 'SetHotspareType', drive_info.Name, ctx, 1, 0)
    lu.assertEquals(ret, false)

    method_misc.test_controller_vendor = function ()
        return true
    end

    drive.RefControllerTypeId = 1
    sml.get_pd_info = function()
        return { hot_spare = 1 }
    end
    local ret = pcall(rpc_service_drive.drive_operate, rpc_service_drive, 'SetHotspareType', drive_info.Name, ctx, 1, 0)
    lu.assertEquals(ret, true)
    lu.assertEquals(drive.HotspareType, 1)
    sml.get_pd_info = function()
        return { hot_spare = 0 }
    end
    ret = pcall(rpc_service_drive.drive_operate, rpc_service_drive, 'SetHotspareType', drive_info.Name, ctx, 0, 0)
    lu.assertEquals(ret, true)
    lu.assertEquals(drive.HotspareType, 0)
    sml.get_pd_info = function()
        return { hot_spare = 2 }
    end
    ret = pcall(rpc_service_drive.drive_operate, rpc_service_drive, 'SetHotspareType', drive_info.Name, ctx, 2, 0)
    lu.assertEquals(ret, true)
    lu.assertEquals(drive.HotspareType, 2)
    sml.get_pd_info = function()
        return { hot_spare = 3 }
    end
    ret = pcall(rpc_service_drive.drive_operate, rpc_service_drive, 'SetHotspareType', drive_info.Name, ctx, 3, 0)
    lu.assertEquals(ret, true)
    lu.assertEquals(drive.HotspareType, 3)

    ret = pcall(rpc_service_drive.drive_operate, rpc_service_drive, 'SetHotspareType', drive_info.Name, ctx, 4, 0)
    lu.assertEquals(ret, false)
end

-- 测试定位状态更新
function TestStorage:test_update_location_indicator_state()
    local drive_info = {
        HddBackplaneStartSlot = 0,
        Presence = 0,
        RelativeSlot = 1,
        ObjectName = 'Drive1',
        LocateLed = 1,
        FaultLed = 0
    }
    self.object_manager.mc:add_object('Drive', drive_info)
    self.bus_monitor_service.on_os_state_changed:emit(true)
    self.tasks:run_task(self.mctp_service.mctp_start_task)
    local drive = self.drive_collection:get_drive(drive_info.Name)
    lu.assertEquals(drive.LocationIndicatorState, 1)

    drive.LocateLed = 0
    drive.on_property_changed:emit('LocateLed', 0)
    lu.assertEquals(drive.LocationIndicatorState, 0)

    drive.LocateLed = 1
    drive.on_property_changed:emit('LocateLed', 1)
    lu.assertEquals(drive.LocationIndicatorState, 1)
end

local function loop_exec_func(func, times, ...)
    for i = 1, times do
        func(...)
    end
end

function TestStorage:test_firmware_status_err_associated_cpld()
    local controller_1 = {
        Id = 0,
        path = 'Controller_1_0001010102',
        CtrlOption1 = 2275649443,
        CtrlOption2 = 32647,
        CtrlOption3 = 0,
        TypeId = 99,
        OOBSupport = 1
    }
    local controller_2 = {
        Id = 1,
        path = 'Controller_2_0001010102',
        CtrlOption1 = 2275649443,
        CtrlOption2 = 32647,
        CtrlOption3 = 0,
        TypeId = 99,
        OOBSupport = 1
    }

    local storageconfig = self.c_storageconfig
    storageconfig.task = nil

    self.bus_monitor_service.on_os_state_changed:emit(true)
    self.object_manager.mc:add_object('Controller', controller_1, 1)

    loop_exec_func(self.tasks.run_task, 15, self.tasks, storageconfig.task)

    -- 检测15次未满足条件
    lu.assertEquals(storageconfig.all_ctrl_loaded, false)

    -- 加一个控制器重新开始计数
    self.object_manager.mc:add_object('Controller', controller_2, 2)

    loop_exec_func(self.tasks.run_task, 15, self.tasks, storageconfig.task)

    -- 检测15次未满足条件
    lu.assertEquals(storageconfig.all_ctrl_loaded, false)

    self.tasks:run_task(storageconfig.task)

    -- 第16次满足条件
    lu.assertEquals(storageconfig.all_ctrl_loaded, true)

    local drive_info = {
        HddBackplaneStartSlot = 0,
        Presence = 1,
        RelativeSlot = 1,
        ObjectName = 'Drive1',
        LocateLed = 0,
        FaultLed = 1
    }
    self.object_manager.mc:add_object('Drive', drive_info)

    -- 由于FaultLed加入了防抖，需要获取5次
    loop_exec_func(self.tasks.run_all_task, 5, self.tasks)
    local drive = self.drive_collection:get_drive('Disk1')
    lu.assertEquals(drive.FirmwareStatusError, true)
    self.object_manager.mc:del_object('Controller', controller_1, 1)
    self.object_manager.mc:del_object('Controller', controller_2, 2)
    self.tasks:run_all_task()
end

function TestStorage:test_update_drive_slot_info()
    local drive_info = { Id = 1, Name = 'Disk1', Presence = 1, ObjectName = 'Drive1' }
    self.object_manager.mc:add_object('Drive', drive_info)
    self.bus_monitor_service.on_os_state_changed:emit(true)
    self.tasks:run_task(self.mctp_service.mctp_start_task)

    local drive = self.drive_collection:get_drive('Disk1')
    lu.assertEquals(drive.Id, 1)
end

-- 测试物理盘dump日志
function TestStorage:test_drive_dump_info()
    -- 添加物理盘对象
    local drive_info, drive = create_drive(self)
    lu.assertNotIsNil(drive)
    lu.assertNotIsNil(drive_info)
    local obj_info_path = project_dir .. '/test/unit/drive_info.txt'
    local fp_w = file_sec.open_s(obj_info_path, 'w+')
    lu.assertNotIsNil(fp_w)

    -- 调用物理盘收集日志到桩文件中
    drive:dump_physical_drive(fp_w)
    fp_w:close()

    local file = file_sec.open_s(obj_info_path, "r")
    lu.assertNotIsNil(file)
    local content = utils.close(file, pcall(file.read, file, "*a"))
    lu.assertNotIsNil(content)
    -- 检查物理盘收集的信息
    lu.assertNotEquals(string.find(content, 'On Hours                           : N/A'), nil)

    utils.remove_file(obj_info_path)
end

-- 测试增加磁盘事件
function TestStorage:test_drive_add_event()
    local event = {
        path = "/bmc/kepler/Systems/1/Events",
        AddEvent_PACKED = function()
            return {
                unpack = function()
                    return 1
                end
            }
        end
    }
    client.ForeachEventsObjects = function(self, cb)
        cb(event)
    end
    local ret = pcall(add_event.generate_drive_max_temp, add_event, 'true', 60)
    lu.assertIsTrue(ret)

    ret = pcall(add_event.generate_percentage_threshold, add_event, 'true', 'Disk0')
    lu.assertEquals(ret, true)
    ret = pcall(add_event.generate_nvme_link_fault, add_event, 'true', 'Disk0')
    lu.assertEquals(ret, true)
end

--测试更新nvme盘数据
function TestStorage:test_update_nvme_mi_info()
    c_nvme.SerialNumber = 'N/A'
    c_nvme.vpd_nvme_mi_get_product_info = function()
        return string.char(140)
    end

    c_nvme:update_static_prop()
    lu.assertEquals(c_nvme.SerialNumber, 'N/A')
    c_nvme.vpd_nvme_mi_get_product_info = function()
        return string.char(120)
    end
    c_nvme:update_static_prop()
    lu.assertEquals(c_nvme.SerialNumber, 'x')

    local ret = pcall(c_nvme.vpd_nvme_mi_get_vender_id, c_nvme, 'INTEL')
    lu.assertIsTrue(ret)

    ret = pcall(c_nvme.vpd_nvme_mi_get_vender_id, c_nvme, 'Samsung')
    lu.assertIsTrue(ret)

    local nvme = {
        SSDChip = 1
    }
    ret = pcall(c_nvme.get_firmeware_version, nvme)
    lu.assertIsTrue(ret)
end

function TestStorage:test_update_dynamic_drive_info()
    c_nvme.TemperatureCelsius = 100
    c_nvme.PredictedMediaLifeLeftPercent = 50
    c_nvme.PowerOnHours = 0
    c_nvme.MediaErrorCount = 0
    c_nvme.Revision = c_nvme:get_firmeware_version() or 'N/A'
    c_nvme.link_fault = false
    c_nvme.Status = 0
    c_nvme.LifeUsedPercentage = 0
    local dynamic_info = c_nvme:update_dynamic_drive_info()
    lu.assertEquals(dynamic_info.Status , 0)
    lu.assertEquals(dynamic_info.LifeUsedPercentage , 0)
end

function TestStorage:test_vpd_nvme_mi_get_info()
    c_nvme.update_static_prop = function(c_nvme)
        c_nvme.nvme_mi_info = {}
        c_nvme.Model = "Huawei"
        c_nvme.SerialNumber = ""
        c_nvme.Manufacturer = ""
        c_nvme.ManufacturerId = 0
        c_nvme.CapableSpeedGbs = 0
        c_nvme.NegotiatedSpeedGbs = 0
        c_nvme.CapacityMiB = 0
        c_nvme.MediaType = 0
        c_nvme.Protocol = 0
        c_nvme.ResourceId = 0
        c_nvme.Status = 0
        c_nvme.LifeUsedPercentage = 0
    end
    local nvme_mi_info = c_nvme:vpd_nvme_mi_get_info()
    lu.assertEquals(nvme_mi_info.Status, 0)
    lu.assertEquals(nvme_mi_info.LifeUsedPercentage, 0)
end

function TestStorage:test_c_nvme_ctor()
    c_nvme:ctor()
    lu.assertEquals(c_nvme.Status, 255)
    lu.assertEquals(c_nvme.LifeUsedPercentage, 255)
end

-- 测试更新nvme盘资源归属
function TestStorage:test_update_nvme_resource_id()
    c_nvme.nvme_info = {}
    c_nvme.RefComponent = {}
    local socket_id = 1
    mdb.get_sub_objects = function(...)
        return {
            {
                SlotID = 0,
                ComponentType = 0,
                SocketID = socket_id - 1
            }
        }
    end

    mdb.get_object = function(...)
        return {
            Instance = 0,
            Type = 0
        }
    end
    c_nvme:update_res_id()
    lu.assertEquals(c_nvme.nvme_info.ResouceId, socket_id)

    socket_id = 0XFF
    c_nvme:update_res_id()
    lu.assertEquals(c_nvme.nvme_info.ResouceId, socket_id - 1)
end

-- 测试返回硬盘慢盘参数
function TestStorage:test_get_drives_io_diagnose_info()
    local _, drive = create_drive(self)
    drive.Name = 'Disk1'
    self.drive_collection.get_all_drives = function ()
        return {drive}
    end

    local ret = pcall(rpc_service_subhealth.get_drives_subhealth_diagnose_info,
        rpc_service_subhealth, 'IODeterioration')
    lu.assertEquals(ret, true)
end

-- 测试获取硬盘寿命信息
function TestStorage:test_get_drives_lifespan_info()
    create_drive(self)
    self.tasks:run_all_task()

    local ok, ret = pcall(rpc_service_subhealth.get_drives_subhealth_diagnose_info,
        rpc_service_subhealth, 'EstimatedLifespan')
    lu.assertEquals(ok, true)
    lu.assertNotIsNil(ret)
end

function TestStorage:test_update_drive_static_info_by_sepecific_obj()
    local _, drive = create_drive(self)
    local obj = {update_drive_info = function()
        local info = {}
        info.Model = "Huawei"
        info.SerialNumber = ""
        info.Manufacturer = ""
        info.ManufacturerId = 0
        info.CapableSpeedGbs = 0
        info.NegotiatedSpeedGbs = 0
        info.CapacityMiB = 0
        info.MediaType = 0
        info.Protocol = 0
        info.ResourceId = 0
        info.Status = 0
        info.LifeUsedPercentage = 0
        return info
    end}
    drive:update_drive_static_info_by_sepecific_obj(obj)
    lu.assertEquals(drive.Status, 0)
    lu.assertEquals(drive.LifeUsedPercentage, 0)
end

--测试更新nvme盘smartlog
function TestStorage:test_update_nvme_smart_info()
    local _, drive = create_drive(self)
    c_nvme.get_nvme_smart_info = function ()
        return {
            avail_spare = 8,
            critical_warning = 1,
            percent_used = 64,
            power_on_hours = 952
        }
    end
    c_nvme.get_nvme_hw_defined_smart_info = function ()
    end
    c_nvme.nvme_mi_mctp_obj = {
        nvme_mi_obj = {}
    }
    drive:update_nvme_smart_log_info({})
    lu.assertEquals(drive.CriticalWarning, 0xff)
    drive:update_nvme_smart_log_info(c_nvme)
    lu.assertEquals(drive.CriticalWarning, 1)
    lu.assertEquals(drive.AvailableSpare, 8)
    lu.assertEquals(drive.TLCSpareBlockPercentage, 8)
    lu.assertEquals(drive.UsedPercentage, 64)
    lu.assertEquals(drive.PredictedMediaLifeLeftPercent, 36)
    lu.assertEquals(drive.PowerOnHours, 952)
end

-- 测试更新nvme盘hwdefined smartlog
function TestStorage:test_update_nvme_hw_defined_smart_info()
    local _, drive = create_drive(self)
    local nvme = {
        get_nvme_hw_defined_smart_info = function ()
            return {
                remnant_wearout = 8,
                slc_avg_ec = 1,
                tlc_avg_ec = 64,
                slc_spare_block = 98,
                tlc_spare_block = 99,
            }
        end,
        nvme_mi_mctp_obj = {
            nvme_mi_obj = {}
        },
        get_nvme_smart_info = function ()
        end
    }

    drive:update_nvme_smart_log_info(nvme)
    lu.assertEquals(drive.PredictedMediaLifeLeftPercent, 8)
    lu.assertEquals(drive.estimated_remaining_lifespan_info.is_support_hw_defined, 1)
    lu.assertEquals(drive.TLCSpareBlockPercentage, 99)
    lu.assertEquals(drive.SLCSpareBlockPercentage, 98)
    lu.assertEquals(drive.estimated_remaining_lifespan_info.slc_avg_ec, 1)
    lu.assertEquals(drive.estimated_remaining_lifespan_info.tlc_avg_ec, 64)
end

-- 测试硬盘冗余信息监控功能
function TestStorage:test_spare_black_to_maintaince_log()
    local _, drive = create_drive(self)
    drive.SLCSpareBlockPercentage = 100
    drive.TLCSpareBlockPercentage = 100

    local spareblock = {
        slc_value = 70,
        tlc_value = 70
    }

    drive.Name = 'Disk1'
    drive.last_record_spare_block = 0
    -- 测试传入无效硬盘id
    local ret = pcall(rpc_service_drive.mock_record_spare_block, rpc_service_drive, 0, 100, 75)
    lu.assertEquals(ret, true)
    -- 测试传入用户区与非用户区冗余信息均越界
    ret = pcall(rpc_service_drive.mock_record_spare_block, rpc_service_drive, 1, 101, 101)
    lu.assertEquals(ret, true)
    -- 测试debug模式及冗余块达到70%
    ret = pcall(rpc_service_drive.mock_record_spare_block, rpc_service_drive, 1, 100, 75)
    lu.assertEquals(ret, true)

    -- 测试正常更新流程及冗余块达到50%
    spareblock.tlc_value = 50
    spareblock.slc_value = 50
    drive:record_maintaince_log_for_spare_black(spareblock, true)

    -- 测试冗余块达到30%
    spareblock.tlc_value = 30
    spareblock.slc_value = 30
    drive:record_maintaince_log_for_spare_black(spareblock, false)

    -- 测试用户区冗余信息更新
    spareblock.tlc_value = 30
    spareblock.slc_value = common_def.INVALID_U8
    drive:record_maintaince_log_for_spare_black(spareblock, false)

    -- 测试24小时内再次更新
    drive.spare_block_last_check_time = math.floor(os.time())
    drive:record_maintaince_log_for_spare_black(spareblock, true)
end

function TestStorage:test_update_smbios_status()
    c_nvme.new = function ()
        c_nvme:ctor()
        c_nvme:init()
    end
    c_nvme.super.init = function ()
    end
    c_nvme.new()
    c_nvme.Slot = 1
    local _, drive = create_drive(self)
    drive.Name = 'Disk1'
    self.drive_collection.get_drive = function ()
        return drive
    end
    self.drive_collection.on_smbios_status_changed:emit(true)
    self.drive_collection.on_add_nvme:emit(c_nvme)
    self.drive_collection.on_smbios_status_changed:emit(true)
    c_nvme.protocol = common_def.NVME_VPD_PROTOCOL_NVME_MI
    self.drive_collection.on_smbios_status_changed:emit(true)
    c_nvme.check_support_mctp = function ()
        return true
    end
    client.GetPCIeDeviceObjects = function ()
        return {}
    end
    self.drive_collection.on_smbios_status_changed:emit(true)
    lu.assertEquals(drive.AvailableSpare, common_def.INVALID_U8)
    lu.assertEquals(drive.CriticalWarning, common_def.INVALID_U8)
    lu.assertEquals(drive.UsedPercentage, common_def.INVALID_U8)
end

function TestStorage:test_dump_log()
    path = '/test/'
    storageApp.drive_collection = {nvme_list={}}
    ret = pcall(storageApp.dump_log, storageApp, nil, path)
    lu.assertEquals(ret, true)
end